package org.huangrui.spark.scala.streaming

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.sql.{Connection, Date, PreparedStatement}
import java.text.SimpleDateFormat
import org.huangrui.spark.scala.util.JDBCUtil
/**
 * @author hr
 * @create 2020-12-27 0:49 
 */
object SparkStreaming12_Req2 {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val sc: StreamingContext = new StreamingContext(sparkConf, Seconds(3))

    //    1.定义 Kafka 参数
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop210:9092,hadoop211:9092,hadoop212:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "spark",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )
    //    2.读取 Kafka 数据创建 DStream
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](sc,
      LocationStrategies.PreferConsistent,//采集的节点和计算的节点该如何做匹配，类似于spark core中的首选位置
      ConsumerStrategies.Subscribe[String, String](Set("spark"), kafkaPara))
    val adclickData: DStream[AdClickData] = kafkaDStream.map {
      kafkaData: ConsumerRecord[String, String] => {
        val data: String = kafkaData.value()
        val datas: Array[String] = data.split(" ")
        AdClickData(datas(0), datas(1), datas(2), datas(3), datas(4))
      }
    }
    //TODO 地区各城市各广告的点击总流量
    val reduceDs: DStream[((String, String, String, String), Int)] = adclickData.map {
      data: AdClickData => {
        val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")
        val day: String = sdf.format(new Date(data.ts.toLong))
        val area: String = data.area
        val city = data.city
        val ad: String = data.ad
        ((day, area, city, ad), 1)
      }
    }.reduceByKey(_ + _)
    reduceDs.foreachRDD{
      rdd=>{
        rdd.foreachPartition{
          iter =>{
            val conn: Connection = JDBCUtil.getConnection
            val pstat: PreparedStatement = conn.prepareStatement(
              """
                | insert into area_city_ad_count ( dt, area, city, adid, count )
                | values ( ?, ?, ?, ?, ? )
                | on DUPLICATE KEY
                | UPDATE count = count + ?
                |""".stripMargin)

            iter.foreach{
              case ((day, area, city, ad), count) =>{
                println(s"${day} ${area} ${city} ${ad} ${count}")
                pstat.setString(1,day)
                pstat.setString(2,area)
                pstat.setString(3,city)
                pstat.setString(4,ad)
                pstat.setInt(5,count)
                pstat.setInt(6,count)
                pstat.executeUpdate()
              }
            }
            pstat.close()
            conn.close()
          }
        }
      }
    }



    sc.start()
    sc.awaitTermination()
  }
  // 广告点击数据
  case class AdClickData( ts:String, area:String, city:String, user:String, ad:String )
}
